package main

import (
	"encoding/json"
	"io"
	"log"
	"net/http"
	"os"
	"strconv"
	"sync"
)

// 关于高并发的简单设计，处理海量数据
// 消费海量的请求
// 在前面的生产者、消费者并发模型中，只有当生产者和消费的速度近似相等时才会达到最佳的效果，同时通过引入带缓存的管道可以消除因临时效率波动产生的影响。但是当生产者和消费者的速度严重不匹配时，我们是无法通过带缓存的管道来提高性能的（缓存的管道只能延缓问题发生的时间，无法消除速度差异带来的问题）。当消费者无法及时消费生产者的输出时，时间积累会导致问题越来越严重。

// 对于生产者、消费者并发模型，我们当然可以通过降低生产者的产能来避免资源的浪费。但在很多场景中，生产者才是核心对象，它们生产出各种问题或任务单据，这时候产出的问题是必须要解决的、任务单据也是必须要完成的。在现实生活中，制造各种生活垃圾的海量人类其实就是垃圾生产者，而清理生活垃圾的少量的清洁工就是垃圾消费者。在网络服务中，提交POST数据的海量用户则变成了生产者，Web后台服务则对应POST数据的消费者。海量生产者的问题也就变成了：如何构造一个能够处理海量请求的Web服务（假设每分钟百万级请求）。

// 在Web服务中，用户提交的每个POST请求可以看作是一个Job任务，而服务器是通过后台的Worker工作者来消费这些Job任务。当面向海量的Job处理时，我们一般可以通过构造一个Worker工作者池来提高Job的处理效率；通过一个带缓存的Job管道来接收新的任务请求，避免任务请求功能无法响应；Job请求接收管道和Worker工作者池通过分发系统来衔接。

// 我们可以用管道来模拟工作者池：当需要处理一个任务时，先从工作者池取一个工作者，处理完任务之后将工作者返回给工作者池。WorkerPool对应工作者池，Worker对应工作者。
// -------------------------------------
// worker pool
// -------------------------------------
type WorkerPool struct {
	workers []*Worker
	pool    chan *Worker
}

// 构造工作池
func NewWorkerPool(maxWorkers int) *WorkerPool {
	p := &WorkerPool{
		workers: make([]*Worker, maxWorkers),
		pool:    make(chan *Worker, maxWorkers),
	}
	// 初始化工作者
	for i, _ := range p.workers {
		worker := NewWorker(0)
		p.workers[i] = worker
		p.pool <- worker
	}
	return p
}

// 启动工作者
func (p *WorkerPool) Start() {
	for _, worker := range p.workers {
		worker.Start()
	}
}

// 停止工作者
func (p *WorkerPool) Stop() {
	for _, worker := range p.workers {
		worker.Stop()
	}
}

// 获取所有工作者<堵塞>
func (p *WorkerPool) Get() *Worker {
	return <-p.pool
}

// 返回工作者
func (p *WorkerPool) Put(w *Worker) {
	p.pool <- w
}

// -------------------------------------
// worker
// -------------------------------------
type Worker struct {
	job  chan interface{}
	quit chan bool
	wg   sync.WaitGroup
}

// 构造工作者
func NewWorker(maxJobs int) *Worker {
	return &Worker{
		job:  make(chan interface{}, maxJobs),
		quit: make(chan bool),
	}
}

// 启动任务
func (w *Worker) Start() {
	w.wg.Add(1)
	go func() {
		defer w.wg.Done()
		for {
			// 接受任务
			// 此时工作中已经从工作者池中取出
			select {
			case job := <-w.job:
				// 处理任务
			case <-w.quit:
				return
			}
		}
	}()
}

// 关闭任务
func (p *Worker) Stop() {
	p.quit <- true
	p.wg.Wait()
}

// 提交任务
func (p *Worker) AddJob(job interface{}) {
	p.job <- job
}

// -------------------------------------
// server 任务分发系统
// -------------------------------------
type Service struct {
	workers *WorkerPool
	jobs    chan interface{}
	maxJobs int
	wg      sync.WaitGroup
}

func NewService(maxWorkers, maxJobs int) *Service {
	return &Service{
		workers: NewWorkerPool(maxWorkers),
		jobs:    make(chan interface{}, maxJobs),
	}
}

func (p *Service) Start() {
	p.jobs = make(chan interface{}, p.maxJobs)
	p.wg.Add(1)
	p.workers.Start()

	go func() {
		defer p.wg.Done()

		for job := range p.jobs {
			go func(job *Worker) {
				// 从工作者池取出一个工作者
				worker := p.workers.Get()
				// 完成任务后返回给工作者池
				defer p.workers.Put(worker)
				// 提交任务处理(异步)
				worker.AddJob(job)

			}(job)
		}
	}()
}

func (p *Service) Stop() {
	p.workers.Stop()
	close(p.jobs)
	p.wg.Wait()
}

// 提交任务
// 任务管道带较大的缓存，延缓阻塞时间
func (p *Service) AddJob(job interface{}) {
	p.jobs <- job
}

// -------------------------------------
// 主程序任务分发系统
// -------------------------------------
var (
	MaxWorker, _ = strconv.Atoi(os.Getenv("MAX_WORKERS"))
	MaxQueue, _  = strconv.Atoi(os.Getenv("MAX_QUEUE"))
)

func main() {
	service := NewService(MaxWorker, MaxQueue)

	service.Start()
	defer service.Stop()
	// 处理海量任务

	http.HandlerFunc("/jobs", func(w http.ResponseWriter, r *http.Request) {
		if r.Method != "POST" {
			w.WriteHeader(http.StatusMethodNotAllowed)
		}
		// Job以json格式提交
		var jobs []Job // JOb有问题 . MaxLength 未定义
		err := json.NewDecoder(io.LimitReader(r.Body, MaxLength)).Decode(&job)
		if err != nil {
			w.Header().Set("Content-Type", "application/json; charset=UTF-8")
			w.WriteHeader(http.StatusBadRequest)
			return
		}

		// 处理任务
		for _, job := range jobs {
			service.AddJob(job)
		}

		// OK
		w.WriteHeader(http.StatusOK)
	})

	// 启动web服务
	log.Fatal(http.ListenAndServe(":8080", nil))
}
